package org.example.kpi;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KPIPV {

	public static class KPIPVMapper extends
			Mapper<Object, Text, Text, IntWritable> {
		private IntWritable one = new IntWritable(1);
		private Text word = new Text();

		/**
		 * 对输入的键值对进行处理，将符合条件的值转换为特定的键值对输出。
		 *
		 * @param key 输入键，代表一行数据的唯一标识。
		 * @param value 输入值，包含需要处理的字符串数据。
		 * @param context 上下文对象，用于输出处理后的键值对。
		 * @throws IOException 当进行I/O操作时发生错误。
		 * @throws InterruptedException 当执行被中断时抛出。
		 */
		@Override
		public void map(Object key, Text value, Context context)
		        throws IOException, InterruptedException {
		    // 从输入值中过滤并解析出KPI对象
		    KPI kpi = KPI.filterPVs(value.toString());
		    // 如果KPI对象有效，则将其请求部分作为新的键，输出到上下文中
		    if (kpi.isValid()) {
		        word.set(kpi.getRequest());
		        context.write(word, one);
		    }
		}
	}

	public static class KPIPVReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

        /**
         * 在Reduce阶段对指定键的所有值进行求和，并将结果写入上下文中。
         *
         * @param key 指定的键，对应一组值将对此键进行操作。
         * @param values 键对应的多个值，为一个可迭代的IntWritable对象集合。
         * @param context 用于输出结果的上下文对象，可以将计算结果写入其中。
         * @throws IOException 如果发生I/O错误。
         * @throws InterruptedException 如果执行被中断。
         */
		@Override
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			int sum = 0; // 初始化求和变量
			for (IntWritable value : values) {
				sum += value.get(); // 遍历并累加所有值
			}
			result.set(sum); // 将求和结果设置到结果对象中
			context.write(key, result); // 将键值对写入上下文
		}

	}

	/**
	 * 主函数：计算网站页面访问量（PV）的关键性能指标（KPI）。
	 * @param args 命令行参数（未使用）
	 * @throws Exception 处理过程中可能抛出的异常
	 */
	public static void main(String[] args) throws Exception {

	    // 定义输入和输出路径
	    String input = "hdfs://192.168.134.132:9000/weblog/logfile";
	    String output = "hdfs://192.168.134.132:9000/weblog/kpi/pv";

	    // 配置Hadoop环境
	    Configuration conf = new Configuration();
	    // 创建并配置作业
	    Job job = Job.getInstance(conf, "KPIPV");
	    job.setJarByClass(KPIPV.class); // 指定作业使用的JAR包
	    // 设置Mapper和Reducer的输出类型
	    job.setMapOutputKeyClass(Text.class);
	    job.setMapOutputValueClass(IntWritable.class);

	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(IntWritable.class);

	    // 设置Mapper和Reducer类
	    job.setMapperClass(KPIPVMapper.class);
	    job.setCombinerClass(KPIPVReducer.class); // 设置Combiner类以减少数据传输量
	    job.setReducerClass(KPIPVReducer.class); // 设置Reducer类进行聚合计算

	    // 添加输入和输出路径
	    FileInputFormat.addInputPath(job, new Path(input));
	    FileOutputFormat.setOutputPath(job, new Path(output));
	    // 提交作业并等待完成
	    if (!job.waitForCompletion(true))
	        return;
	}

}
